package com.atguigu.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.function.DimAsyncFunction;
import com.atguigu.bean.OrderDetail;
import com.atguigu.bean.OrderInfo;
import com.atguigu.bean.OrderWide;
import com.atguigu.common.DWMStaticConstants;
import com.atguigu.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.concurrent.TimeUnit;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 * mysql--flinkcdc--kafka--basedbapp--kafka/hbase--orderwidwapp
 * msyql-flink-kafka-hbase-zookeeper-hadoop--phoenix
 */
public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /**
         * 生产环境下一定要写的，不过学习阶段还要开启hdfs比较麻烦先关闭
         *         System.setProperty("HADOOP_USER_NAME","atguigu");
         *         env.setStateBackend(new FsStateBackend(OdsStaticConstants.CHECK_POINT_URL));
         *         // 设置checkpoint时间
         *         env.enableCheckpointing(5000L);
         *         // 设置精准一次性
         *         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
         *         // 设置check超时时间
         *         env.getCheckpointConfig().setCheckpointTimeout(10000L);
         *         // 设置最多两个checkpoint一块
         *         env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
         *         // 设置两个check间隔
         *         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
         */
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        DataStreamSource<String> orderInfoDs =
                env.addSource(KafkaUtil.getKafkaConsumer(DWMStaticConstants.DWM_TOPIC_ORDER_INFO, DWMStaticConstants.DWM_ORDER_WIDE_GROUP_ID));
        SingleOutputStreamOperator<OrderInfo> orderInfoSingleOutputStreamOperator = orderInfoDs.map(f -> {
            OrderInfo orderInfo = JSON.parseObject(f, OrderInfo.class);
            String[] split = orderInfo.getCreate_time().split(" ");
            orderInfo.setCreate_date(split[0]);
            orderInfo.setCreate_hour(split[1].split(":")[0]);
            orderInfo.setCreate_ts(dateFormat.parse(orderInfo.getCreate_time()).getTime());
            return orderInfo;
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<OrderInfo>forMonotonousTimestamps()
                .withTimestampAssigner((element, recordTimestamp) -> element.getCreate_ts()));

        DataStreamSource<String> orderDetailInfoDs =
                env.addSource(KafkaUtil.getKafkaConsumer(DWMStaticConstants.DWM_TOPIC_ORDER_DETAIL, DWMStaticConstants.DWM_ORDER_WIDE_GROUP_ID));
        SingleOutputStreamOperator<OrderDetail> orderDetailSingleOutputStreamOperator = orderDetailInfoDs.map(f -> {
            OrderDetail orderDetail = JSON.parseObject(f, OrderDetail.class);
            orderDetail.setCreate_ts(dateFormat.parse(orderDetail.getCreate_time()).getTime());
            return orderDetail;
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<OrderDetail>forMonotonousTimestamps()
                .withTimestampAssigner((element, recordTimestamp) -> element.getCreate_ts()));
        // 双流join
        SingleOutputStreamOperator<OrderWide> orderInfoJoinOrderDetailDs = orderInfoSingleOutputStreamOperator.keyBy(OrderInfo::getId)
                .intervalJoin(orderDetailSingleOutputStreamOperator.keyBy(OrderDetail::getOrder_id))
                // 生产环境设置最大延迟时间
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, Context context, Collector<OrderWide> collector) throws Exception {
                        collector.collect(new OrderWide(orderInfo, orderDetail));
                    }
                });
        // 关联用户维度信息
        SingleOutputStreamOperator<OrderWide> orderWideSingleOutputStreamOperatorWithUser =
                AsyncDataStream.unorderedWait(
                        orderInfoJoinOrderDetailDs,
                        new DimAsyncFunction<OrderWide>("DIM_USER_INFO") {
                            @Override
                            public String getId(OrderWide input) {
                                return input.getUser_id().toString();
                            }

                            @Override
                            public void join(JSONObject jsonObject, OrderWide orderWide) throws ParseException {
                                orderWide.setUser_gender(jsonObject.getString("GENDER"));
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                                long birthday = sdf.parse(jsonObject.getString("BIRTHDAY")).getTime();
                                long currentTime = System.currentTimeMillis();
                                long l = currentTime - birthday;
                                long old = l / (1000 * 60 * 60 * 24 * 365L);
                                orderWide.setUser_age(((int) old));
                            }

                        },// zookeeper是60s，因此设置为60s，访问hbase首先访问zookeeper
                        60L, TimeUnit.SECONDS);
        // 地区维度数据
        SingleOutputStreamOperator<OrderWide> orderWideSingleOutputStreamOperatorWithProvince = AsyncDataStream.unorderedWait(
                orderWideSingleOutputStreamOperatorWithUser,
                new DimAsyncFunction<OrderWide>("DIM_BASE_PROVINCE") {
                    @Override
                    public String getId(OrderWide input) {
                        return input.getProvince_id().toString();
                    }

                    @Override
                    public void join(JSONObject jsonObject, OrderWide orderWide) throws ParseException {
                        orderWide.setProvince_name(jsonObject.getString("NAME"));
                        orderWide.setProvince_area_code(jsonObject.getString("AREA_CODE"));
                        orderWide.setProvince_3166_2_code(jsonObject.getString("ISO_3166_2"));
                        orderWide.setProvince_iso_code(jsonObject.getString("ISO_CODE"));
                    }
                },
                60L,
                TimeUnit.SECONDS);
        SingleOutputStreamOperator<OrderWide> orderWideSingleOutputStreamOperatorWithSku = AsyncDataStream.unorderedWait(
                orderWideSingleOutputStreamOperatorWithProvince,
                new DimAsyncFunction<OrderWide>("DIM_SKU_INFO") {
                    @Override
                    public String getId(OrderWide input) {
                        return input.getSku_id().toString();
                    }

                    @Override
                    public void join(JSONObject jsonObject, OrderWide orderWide) throws ParseException {
                        orderWide.setSpu_id(jsonObject.getLong("SPU_ID"));
                        orderWide.setSku_name(jsonObject.getString("SKU_NAME"));
                        orderWide.setCategory3_id(jsonObject.getLong("CATEGORY3_ID"));
                        orderWide.setTm_id(jsonObject.getLong("TN_ID"));
                    }
                },
                60L, TimeUnit.SECONDS);
        SingleOutputStreamOperator<OrderWide> orderWideSingleOutputStreamOperatorWithSpu = AsyncDataStream.unorderedWait(
                orderWideSingleOutputStreamOperatorWithSku,
                new DimAsyncFunction<OrderWide>("DIM_SPU_INFO") {
                    @Override
                    public String getId(OrderWide input) {
                        return input.getSpu_id().toString();
                    }

                    @Override
                    public void join(JSONObject jsonObject, OrderWide orderWide) throws ParseException {
                        orderWide.setSpu_name(jsonObject.getString("SPU_NAME"));
                    }
                },
                60L,
                TimeUnit.SECONDS
        );
        SingleOutputStreamOperator<OrderWide> orderWideSingleOutputStreamOperatorWith3C = AsyncDataStream.unorderedWait(
                orderWideSingleOutputStreamOperatorWithSpu,
                new DimAsyncFunction<OrderWide>("DIM_BASE_CATEGORY3") {
                    @Override
                    public String getId(OrderWide input) {
                        return input.getCategory3_id().toString();
                    }

                    @Override
                    public void join(JSONObject jsonObject, OrderWide orderWide) throws ParseException {
                        orderWide.setCategory3_name(jsonObject.getString("NAME"));
                    }
                },
                60L,
                TimeUnit.SECONDS
        );
        orderWideSingleOutputStreamOperatorWith3C
                .map(JSONObject::toJSONString)
                .addSink(KafkaUtil.getKafkaProducter(DWMStaticConstants.DWM_TOPIC_ORDER_WIDE_SINK));
//        orderInfoJoinOrderDetailDs.print(">>>>>>>>>>join");
        orderWideSingleOutputStreamOperatorWith3C.print("double stream join>>>>>");
        //double stream join>>>>>> OrderWide{detail_id=79580, order_id=26541, sku_id=12,
        // order_price=9197.00, sku_num=2,
        // sku_name='Apple iPhone 12 (A2404) 128GB 黑色 支持移动联通电信5G 双卡双待手机',
        // province_id=33, order_status='1001', user_id=2746, total_amount=24336.00,
        // activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=24330.00,
        // feight_fee=6.00, split_feight_fee=null, split_activity_amount=null,
        // split_coupon_amount=null, split_total_amount=18394.00, expire_time='null',
        // create_time='2021-11-21 23:10:35', operate_time='null', create_date='2021-11-21',
        // create_hour='23', province_name='云南', province_area_code='530000',
        // province_iso_code='CN-53', province_3166_2_code='CN-YN', user_age=44,
        // user_gender='F', spu_id=3, tm_id=null, category3_id=61, spu_name='Apple iPhone 12',
        // tm_name='null', category3_name='手机'}
        env.execute();
    }
}
